package com.mgface.zookeeper.premitive;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.DefaultACLProvider;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class 工作进度汇报 {

	public static void main(String[] args) throws Exception {
		CuratorFramework client = CuratorFrameworkFactory.builder()
				.connectString("localhost:2181,localhost:2182,localhost:2183").sessionTimeoutMs(10_000)
				.connectionTimeoutMs(10_000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
		client.start();
		System.out.println("连接启动完毕");
		String path = "/task_process";
		Stat stat = client.checkExists().forPath(path);
		if (stat != null) {
			client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
		}
		String t1 = ZKPaths.makePath(path, "t1");
		String t2 = ZKPaths.makePath(path, "t2");
		List<String> list = new ArrayList<>();
		list.add(t2);
		list.add(t1);
		ZooKeeper zookeeper = client.getZookeeperClient().getZooKeeper();

		for (String s : list) {
			ZKPaths.mkdirs(zookeeper, s, true, new DefaultACLProvider());
		}
		List<PathChildrenCache> caches = new ArrayList<>();
		System.out.println(client.getChildren().forPath(path));
		for (String s : list) {
			//监控子节点
			PathChildrenCache cache = new PathChildrenCache(client, s, true);
			caches.add(cache);
			cache.start(StartMode.POST_INITIALIZED_EVENT);
			cache.getListenable().addListener((clent, event) -> {
				switch (event.getType()) {
				case CHILD_ADDED:
					System.out.println("节点-->添加:" + event.getData().getPath() + ",child_data:"+ new String(event.getData().getData()));
					break;
				case CHILD_REMOVED:
					System.out.println("节点-->删除:" + event.getData().getPath() + ",完成任务。");
					break;
				case CHILD_UPDATED:
					System.out.println("节点-->修改:" + event.getData().getPath() + ",child_data:"+ new String(event.getData().getData()));
					break;
				default:
					break;
				}
			});
		}
		CountDownLatch cdl = new CountDownLatch(1);
		Thread td1 = new Thread(() -> {
			try {
				cdl.await();
				String newPath = client.create().withMode(CreateMode.EPHEMERAL).forPath(t1+"/task1");
				//拿到父节点的值做一系列的操作
				//String v = new String(client.getData().forPath(t1));
				//回写二次进度报告
				client.setData().forPath(newPath, ("process>>>"+"50%").getBytes());
				TimeUnit.SECONDS.sleep(15);
				client.setData().forPath(newPath, ("process>>>"+"100%").getBytes());
				client.delete().guaranteed().forPath(newPath);
			} catch (Exception e) {
				e.printStackTrace();
			}
		});
		td1.start();
		Thread td2 = new Thread(() -> {
			try {
				cdl.await();
				String newPath = client.create().withMode(CreateMode.EPHEMERAL).forPath(t2+"/task2");
				//拿到父节点的值做一系列的操作
				//String v = new String(client.getData().forPath(t2));
				//回写二次进度报告
				client.setData().forPath(newPath, ("process>>>"+"50%").getBytes());
				TimeUnit.SECONDS.sleep(15);
				client.setData().forPath(newPath, ("process>>>"+"75%").getBytes());
				TimeUnit.SECONDS.sleep(15);
				client.setData().forPath(newPath, ("process>>>"+"100%").getBytes());
				client.delete().guaranteed().forPath(newPath);
			} catch (Exception e) {
				e.printStackTrace();
			}
		});
		td2.start();
		TimeUnit.SECONDS.sleep(10);
		cdl.countDown();
		TimeUnit.SECONDS.sleep(90);
		for (PathChildrenCache cache : caches) {
			cache.close();
		}
	}

}
